package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.UserJumpBean;
import com.atguigu.edu.realtime.util.ClickHouseUtil;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import scala.Tuple4;

import java.time.Duration;

public class DwsTrafficUserJumpWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // TODO 3 读取流的数据
        String ujTopic = "dwd_traffic_user_jump_detail";
        String groupID = "dws_traffic_user_jump_window";
        DataStreamSource<String> ujStream = env.addSource(KafkaUtil.getKafkaConsumer(ujTopic, groupID));

        // TODO 4 转换为相同的结构javaBean
        SingleOutputStreamOperator<UserJumpBean> ujBeanStream = ujStream.map(new MapFunction<String, UserJumpBean>() {
            @Override
            public UserJumpBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                return new UserJumpBean("", "", common.getString("is_new"), common.getString("sc"),
                         1L,jsonObject.getLong("ts"));
            }
        });

        // TODO 5 设置水位线
        SingleOutputStreamOperator<UserJumpBean> beanWithWatermarkStream = ujBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<UserJumpBean>forBoundedOutOfOrderness(Duration.ofSeconds(15)).withTimestampAssigner(new SerializableTimestampAssigner<UserJumpBean>() {
            @Override
            public long extractTimestamp(UserJumpBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        // TODO 6 开窗聚合数据
        KeyedStream<UserJumpBean, Tuple2<String, String>> keyedStream = beanWithWatermarkStream.keyBy(new KeySelector<UserJumpBean, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(UserJumpBean value) throws Exception {
                return new Tuple2<>(value.getIsNew(),value.getSource());
            }
        });

        WindowedStream<UserJumpBean, Tuple2<String, String>, TimeWindow> windowStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10))).allowedLateness(Time.seconds(2));

        SingleOutputStreamOperator<UserJumpBean> reduceStream = windowStream.reduce(new ReduceFunction<UserJumpBean>() {
            @Override
            public UserJumpBean reduce(UserJumpBean value1, UserJumpBean value2) throws Exception {
                value1.setSession_out_num(value1.getSession_out_num() + value2.getSession_out_num());
                return value1;
            }
        }, new WindowFunction<UserJumpBean, UserJumpBean, Tuple2<String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple2<String, String> stringString, TimeWindow window, Iterable<UserJumpBean> input, Collector<UserJumpBean> out) throws Exception {
                UserJumpBean UserJumpBean = input.iterator().next();
                UserJumpBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                UserJumpBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                UserJumpBean.setTs(System.currentTimeMillis());
                out.collect(UserJumpBean);
            }
        });

//        reduceStream.print();


        // TODO 7 写出到clickHouse
        reduceStream.addSink(ClickHouseUtil.<UserJumpBean>getClickHouseSinkFunc("insert into dws_traffic_user_jump_window values(?,?,?,?,?,?)"));

        // TODO 8 执行任务
        env.execute(groupID);
    }
}
